package com.hkbigdata.processapi;

import com.hkbigdata.bean.WaterSensor;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.IterationRuntimeContext;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @author liuanbo
 * @creat 2024-05-14-17:35
 * @see 2194550857@qq.com
 */
public class Flink01_ProcessFunction {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);

        SingleOutputStreamOperator<WaterSensor> waterSensorSingleOutputStreamOperator = env.socketTextStream("localhost", 9999)
                .flatMap(new FlatMapFunction<String, WaterSensor>() {
                    @Override
                    public void flatMap(String value, Collector<WaterSensor> out) throws Exception {
                        String[] split = value.split(",");
                        out.collect(new WaterSensor(
                                split[0],
                                Long.valueOf(split[1]),
                                Double.valueOf(split[2])
                        ));
                    }
                });

        //1.如果是接口，实现接口的话，那就需要重写接口里面所有方法（FlatMapFunction）
        //2.如果说是抽象类，继承抽象类，只要需要重写所有抽象方法（ProcessFunction）

        //第一层 ：匿名内部类 自定义类
        //第二层：ProcessFunction 直接继承抽象类 AbstractRichFunction 包含父类所有方法+ontimer 定时器和output侧输出流+状态编程
        //第三层：AbstractRichFunction 直接实现接口 RichFunction，Serializable open() 生命周期方法和RuntimeContext运行时上下文访问。
        //第四层 ：RichFunction 继承 Function
        //第五层 ：Function 继承 Serializable
        waterSensorSingleOutputStreamOperator.process(new ProcessFunction<WaterSensor, String>() {
            //重写抽象类的抽象方法
            @Override
            public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {
                ctx.output(new OutputTag<String>("id") {
                }, value.toString());
                ctx.timerService();
                out.collect(value.toString());
            }

            @Override
            public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
                super.onTimer(timestamp, ctx, out);
            }

            @Override
            public void setRuntimeContext(RuntimeContext t) {
                super.setRuntimeContext(t);
            }

            @Override
            public RuntimeContext getRuntimeContext() {
                return super.getRuntimeContext();
            }

            @Override
            public IterationRuntimeContext getIterationRuntimeContext() {
                return super.getIterationRuntimeContext();
            }

            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
            }

            @Override
            public void close() throws Exception {
                super.close();
            }
        }).print();

        env.execute();
    }
}
